package com.sui.bigdata.rtcadmin.util;

import com.sui.bigdata.rtcadmin.model.dto.SparkAppDto;
import com.sui.bigdata.rtcadmin.repository.model.SparkApp;
import org.apache.commons.lang.StringUtils;
import org.apache.spark.launcher.SparkLauncher;

/**
 * @Author: TingWuHuang
 * @Date: 2020/5/6 15:30
 * @description
 */
public class SparkArgumentsUtil {

    /**
     *  为spark添加额外的参数，多个参数使用逗号隔开
     *  如：spark.driver.memory=1g,spark.executor.cores=2
     * @param sparkLauncher
     * @param extendParam
     * @return
     */
    public static void addextendParam(SparkLauncher sparkLauncher, String extendParam){
        if (StringUtils.isEmpty(extendParam)){
            return;
        }
        String[] args = extendParam.split(",");
        for(String arg: args){
            String[] keyValue = arg.split("=", 2);
            sparkLauncher.setConf(keyValue[0], keyValue[1]);
        }
    }

    /**
     * 组装application restart的参数
     * @param currentConf
     * @param oldConf
     * @return
     */
    public static SparkAppDto buildRestartConf(SparkAppDto currentConf, SparkApp oldConf){

        if(StringUtils.isNotEmpty(currentConf.getModel())){
            oldConf.setModel(currentConf.getModel());
        }
        if(StringUtils.isNotEmpty(currentConf.getAppResource())){
            oldConf.setAppResource(currentConf.getAppResource());
        }
        if(StringUtils.isNotEmpty(currentConf.getAppArgs())){
            oldConf.setAppArgs(currentConf.getAppArgs());
        }
        if(StringUtils.isNotEmpty(currentConf.getDriverMemory())){
            oldConf.setDriverMemory(currentConf.getDriverMemory());
        }
        if(StringUtils.isNotEmpty(currentConf.getDriverJavaOption())){
            oldConf.setDriverJavaOption(currentConf.getDriverJavaOption());
        }
        if(StringUtils.isNotEmpty(currentConf.getDeveloper())){
            oldConf.setDeveloper(currentConf.getDeveloper());
        }
        if(StringUtils.isNotEmpty(currentConf.getExecutorMemory())){
            oldConf.setExecutorMemory(currentConf.getExecutorMemory());
        }
        if(StringUtils.isNotEmpty(currentConf.getExecutorJavaOption())){
            oldConf.setExecutorJavaOption(currentConf.getExecutorJavaOption());
        }
        if(StringUtils.isNotEmpty(currentConf.getExecutorCores())){
            oldConf.setExecutorCores(currentConf.getExecutorCores());
        }
        if(StringUtils.isNotEmpty(currentConf.getNumExecutors())){
            oldConf.setNumExecutors(currentConf.getNumExecutors());
        }
        if(StringUtils.isNotEmpty(currentConf.getExtendParam())){
            oldConf.setExtendParam(currentConf.getExtendParam());
        }
        if(StringUtils.isNotEmpty(currentConf.getMainClass())){
            oldConf.setMainClass(currentConf.getMainClass());
        }
        if(StringUtils.isNotEmpty(currentConf.getQueue())){
            oldConf.setQueue(currentConf.getQueue());
        }
        if(StringUtils.isNotEmpty(currentConf.getAlarmGroupPhone())) {
            oldConf.setAlarmGroupPhone(currentConf.getAlarmGroupPhone());
        }
        if(StringUtils.isNotEmpty(currentConf.getRemarks())){
            oldConf.setRemarks(currentConf.getRemarks());
        }
        return SparkAppDto.buildSparkAppDto(oldConf);
    }
}
